package com.zoe.flume.sink.elasticsearch;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Event;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Created by shenyuhang on 2018/2/1.
 */
public class ElasticSearchEventSerializer {

    private static final Logger logger = LoggerFactory.getLogger(ElasticSearchEventSerializer.class);

    private XContentBuilder builder;

    public ElasticSearchEventSerializer() {
        try {
            builder = XContentFactory.jsonBuilder().startObject();
        } catch (IOException e) {
            logger.info("ElasticSearchEventSerializer() error", e);
        }
    }

    public XContentBuilder serializer(Event event) throws Exception {
        String body = new String(event.getBody());
        if (Strings.isNullOrEmpty(body)) return null;
        JSONObject json = null;
        try {
            json = JSON.parseObject(body);
            for (String key : json.keySet()) {
                builder.field(key, json.get(key));
            }
        } catch (JSONException ex) {
            builder.field("content", body);
        }
        return builder.endObject();
    }
}
